package bma.common.langutil.concurrent;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

public class BMAThreadPoolExecutorFactoryBean extends CustomizableThreadFactory
		implements BeanNameAware, InitializingBean, DisposableBean,
		FactoryBean<ExecutorService> {

	private static final long serialVersionUID = 1L;

	protected final Log logger = LogFactory.getLog(getClass());

	private int corePoolSize = 1;

	private int maxPoolSize = 10;

	private int keepAliveSeconds = 60;

	private boolean allowCoreThreadTimeOut = false;

	private int queueCapacity = Integer.MAX_VALUE;

	private boolean exposeUnconfigurableExecutor = false;

	private ExecutorService exposedExecutor;

	private ThreadFactory threadFactory = this;

	private boolean threadNamePrefixSet = false;

	private BMARejectedExecutionHandler rejectedExecutionHandler = new BMAThreadPoolExecutor.AbortPolicy();

	private boolean waitForTasksToCompleteOnShutdown = true;
	private long waitForTasksToCompleteTime = 1000;

	private String beanName;

	private boolean prestart = true;

	private ExecutorService executor;

	public void setPrestart(boolean prestart) {
		this.prestart = prestart;
	}

	/**
	 * Set the ThreadPoolExecutor's core pool size. Default is 1.
	 * <p>
	 * <b>This setting can be modified at runtime, for example through JMX.</b>
	 */
	public void setCorePoolSize(int corePoolSize) {
		this.corePoolSize = corePoolSize;
	}

	/**
	 * Set the ThreadPoolExecutor's maximum pool size. Default is
	 * <code>Integer.MAX_VALUE</code>.
	 * <p>
	 * <b>This setting can be modified at runtime, for example through JMX.</b>
	 */
	public void setMaxPoolSize(int maxPoolSize) {
		this.maxPoolSize = maxPoolSize;
	}

	/**
	 * Set the ThreadPoolExecutor's keep-alive seconds. Default is 60.
	 * <p>
	 * <b>This setting can be modified at runtime, for example through JMX.</b>
	 */
	public void setKeepAliveSeconds(int keepAliveSeconds) {
		this.keepAliveSeconds = keepAliveSeconds;
	}

	/**
	 * Specify whether to allow core threads to time out. This enables dynamic
	 * growing and shrinking even in combination with a non-zero queue (since
	 * the max pool size will only grow once the queue is full).
	 * <p>
	 * Default is "false". Note that this feature is only available on Java 6 or
	 * above. On Java 5, consider switching to the backport-concurrent version
	 * of ThreadPoolTaskExecutor which also supports this feature.
	 * 
	 * @see java.util.concurrent.ThreadPoolExecutor#allowCoreThreadTimeOut(boolean)
	 */
	public void setAllowCoreThreadTimeOut(boolean allowCoreThreadTimeOut) {
		this.allowCoreThreadTimeOut = allowCoreThreadTimeOut;
	}

	/**
	 * Set the capacity for the ThreadPoolExecutor's BlockingQueue. Default is
	 * <code>Integer.MAX_VALUE</code>.
	 * <p>
	 * Any positive value will lead to a LinkedBlockingQueue instance; any other
	 * value will lead to a SynchronousQueue instance.
	 * 
	 * @see java.util.concurrent.LinkedBlockingQueue
	 * @see java.util.concurrent.SynchronousQueue
	 */
	public void setQueueCapacity(int queueCapacity) {
		this.queueCapacity = queueCapacity;
	}

	/**
	 * Specify whether this FactoryBean should expose an unconfigurable
	 * decorator for the created executor.
	 * <p>
	 * Default is "false", exposing the raw executor as bean reference. Switch
	 * this flag to "true" to strictly prevent clients from modifying the
	 * executor's configuration.
	 * 
	 * @see java.util.concurrent.Executors#unconfigurableScheduledExecutorService
	 */
	public void setExposeUnconfigurableExecutor(
			boolean exposeUnconfigurableExecutor) {
		this.exposeUnconfigurableExecutor = exposeUnconfigurableExecutor;
	}

	protected ExecutorService initializeExecutor(ThreadFactory threadFactory,
			BMARejectedExecutionHandler rejectedExecutionHandler) {

		BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);
		BMAThreadPoolExecutor executor = new BMAThreadPoolExecutor(
				this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds,
				TimeUnit.SECONDS, queue, threadFactory,
				rejectedExecutionHandler);
		if (this.allowCoreThreadTimeOut) {
			executor.allowCoreThreadTimeOut(true);
		}
		if (prestart) {
			executor.prestartAllCoreThreads();
		}

		ThreadPoolSquare.add(beanName, executor);

		// Wrap executor with an unconfigurable decorator.
		this.exposedExecutor = (this.exposeUnconfigurableExecutor ? Executors
				.unconfigurableExecutorService(executor) : executor);

		return executor;
	}

	/**
	 * Create the BlockingQueue to use for the ThreadPoolExecutor.
	 * <p>
	 * A LinkedBlockingQueue instance will be created for a positive capacity
	 * value; a SynchronousQueue else.
	 * 
	 * @param queueCapacity
	 *            the specified queue capacity
	 * @return the BlockingQueue instance
	 * @see java.util.concurrent.LinkedBlockingQueue
	 * @see java.util.concurrent.SynchronousQueue
	 */
	protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
		if (queueCapacity > 0) {
			return new LinkedBlockingQueue<Runnable>(queueCapacity);
		} else {
			return new SynchronousQueue<Runnable>();
		}
	}

	public ExecutorService getObject() throws Exception {
		return this.exposedExecutor;
	}

	public Class<? extends ExecutorService> getObjectType() {
		return (this.exposedExecutor != null ? this.exposedExecutor.getClass()
				: ExecutorService.class);
	}

	public boolean isSingleton() {
		return true;
	}

	/**
	 * Set the ThreadFactory to use for the ThreadPoolExecutor's thread pool.
	 * Default is the ThreadPoolExecutor's default thread factory.
	 * 
	 * @see java.util.concurrent.Executors#defaultThreadFactory()
	 */
	public void setThreadFactory(ThreadFactory threadFactory) {
		this.threadFactory = (threadFactory != null ? threadFactory : this);
	}

	@Override
	public void setThreadNamePrefix(String threadNamePrefix) {
		super.setThreadNamePrefix(threadNamePrefix);
		this.threadNamePrefixSet = true;
	}

	/**
	 * Set the RejectedExecutionHandler to use for the ThreadPoolExecutor.
	 * Default is the ThreadPoolExecutor's default abort policy.
	 * 
	 * @see java.util.concurrent.ThreadPoolExecutor.AbortPolicy
	 */
	public void setRejectedExecutionHandler(
			BMARejectedExecutionHandler rejectedExecutionHandler) {
		this.rejectedExecutionHandler = (rejectedExecutionHandler != null ? rejectedExecutionHandler
				: new BMAThreadPoolExecutor.AbortPolicy());
	}

	/**
	 * Set whether to wait for scheduled tasks to complete on shutdown.
	 * <p>
	 * Default is "false". Switch this to "true" if you prefer fully completed
	 * tasks at the expense of a longer shutdown phase.
	 * 
	 * @see java.util.concurrent.ExecutorService#shutdown()
	 * @see java.util.concurrent.ExecutorService#shutdownNow()
	 */
	public void setWaitForTasksToCompleteOnShutdown(
			boolean waitForJobsToCompleteOnShutdown) {
		this.waitForTasksToCompleteOnShutdown = waitForJobsToCompleteOnShutdown;
	}

	public void setWaitForTasksToCompleteTime(long waitForTasksToCompleteTime) {
		this.waitForTasksToCompleteTime = waitForTasksToCompleteTime;
	}

	public void setBeanName(String name) {
		this.beanName = name;
	}

	/**
	 * Calls <code>initialize()</code> after the container applied all property
	 * values.
	 * 
	 * @see #initialize()
	 */
	public void afterPropertiesSet() {
		initialize();
	}

	/**
	 * Set up the ExecutorService.
	 */
	public void initialize() {
		if (logger.isInfoEnabled()) {
			logger.info("Initializing ExecutorService "
					+ (this.beanName != null ? " '" + this.beanName + "'" : ""));
		}
		if (!this.threadNamePrefixSet && this.beanName != null) {
			setThreadNamePrefix(this.beanName + "-");
		}
		this.executor = initializeExecutor(this.threadFactory,
				this.rejectedExecutionHandler);
	}

	/**
	 * Calls <code>shutdown</code> when the BeanFactory destroys the task
	 * executor instance.
	 * 
	 * @see #shutdown()
	 */
	public void destroy() {
		shutdown();
	}

	/**
	 * Perform a shutdown on the ThreadPoolExecutor.
	 * 
	 * @see java.util.concurrent.ExecutorService#shutdown()
	 */
	public void shutdown() {
		if (logger.isInfoEnabled()) {
			logger.info("Shutting down ExecutorService"
					+ (this.beanName != null ? " '" + this.beanName + "'" : ""));
		}
		if (this.waitForTasksToCompleteOnShutdown) {
			this.executor.shutdown();
			try {
				this.executor.awaitTermination(waitForTasksToCompleteTime,
						TimeUnit.MILLISECONDS);
			} catch (InterruptedException e) {
			}
		} else {
			this.executor.shutdownNow();
		}
	}
}
